package com.atguigu.day07;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.shaded.curator4.com.google.common.hash.BloomFilter;
import org.apache.flink.shaded.curator4.com.google.common.hash.Funnels;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.HashSet;


/**
 * 处理一小时的独立访客
 * 使用布隆过滤器
 */
public class Example06 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .readTextFile("D:\\WebSite\\flink\\flink\\java_flink\\src\\main\\resources\\UserBehavior.csv")
                .map(
                        x -> UserBehavior.of(x.split(",")[0], x.split(",")[1], x.split(",")[2], x.split(",")[3], x.split(",")[4])
                )
                .filter(x->x.behavior.equals("pv"))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<UserBehavior>forMonotonousTimestamps()
                        .withTimestampAssigner((SerializableTimestampAssigner<UserBehavior>) (element,ctx)->element.timestamp)
                )
                .keyBy(x->true)
                .window(TumblingEventTimeWindows.of(Time.hours(1)))
                .aggregate(new CustomerACC(),new CustomerProcess())
                .print();
        env.execute();
    }

    public static class CustomerProcess extends ProcessWindowFunction<Long,String,Boolean, TimeWindow>{

        @Override
        public void process(Boolean aBoolean, Context context, Iterable<Long> elements, Collector<String> out) throws Exception {
            out.collect("窗口 "+context.window().getStart() +"~"+context.window().getEnd()+"一小时的用户访问量为："+elements.iterator().next());
        }
    }

    public static class CustomerACC implements AggregateFunction<UserBehavior, Tuple2<BloomFilter<Long>, Long>, Long> {
        @Override
        public Tuple2<BloomFilter<Long>, Long> createAccumulator() {
            return Tuple2.of(BloomFilter.create(Funnels.longFunnel(),10000L,0.0001),0L);
        }

        @Override
        public Tuple2<BloomFilter<Long>, Long> add(UserBehavior value, Tuple2<BloomFilter<Long>, Long> accumulator) {
            //如果布隆过滤器中不包含
            if (!accumulator.f0.mightContain(Long.parseLong(value.userId))){
                //将userId加入布隆过滤器  用户数量+1
                accumulator.f0.put(Long.parseLong(value.userId));
                accumulator.f1+=1;
                return accumulator;
            }else {
                //直接返回过滤器
                return accumulator;
            }
        }

        @Override
        public Long getResult(Tuple2<BloomFilter<Long>, Long> accumulator) {
            return accumulator.f1;
        }

        @Override
        public Tuple2<BloomFilter<Long>, Long> merge(Tuple2<BloomFilter<Long>, Long> a, Tuple2<BloomFilter<Long>, Long> b) {
            return null;
        }
    }
    public static class UserBehavior {
        public String userId;
        public String itemId;
        public String categoryId;
        public String behavior;
        public Long timestamp;

        public UserBehavior() {
        }

        public static UserBehavior of(String userId, String itemId, String categoryId, String behavior, String timestamp) {
            return new UserBehavior(userId, itemId, categoryId, behavior, Long.parseLong(timestamp) * 1000L);
        }

        public UserBehavior(String userId, String itemId, String categoryId, String behavior, Long timestamp) {
            this.userId = userId;
            this.itemId = itemId;
            this.categoryId = categoryId;
            this.behavior = behavior;
            this.timestamp = timestamp;
        }
    }
}
